package com.example.kafkastudy.Product;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.HashMap;
import java.util.Map;


public class FirstkafkaProducter {
    public static void main(String[] args){
        // TODO 配置属性集合，用于存放Kafka生产者（消息发送者）的配置信息---连接Kafka集群9092，将消息的Key和Value序列化，
        Map<String, Object> configMap = new HashMap<>();
        // TODO 配置属性：Kafka服务器集群地址
        configMap.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        // TODO 配置属性：Kafka生产的数据为键值对，消息通过网络传输，因此需要先将消息序列化，所以在生产数据进行传输前需要分别对K,V进行对应的序列化操作
//        kafka提供自带的序列化类："org.apache.kafka.common.serialization.StringSerializer"直接传入即可
        //对key进行序列化，常量ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG代表key的序列化类
        configMap.put(
                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");
        configMap.put(
                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringSerializer");

        // TODO 根据配置属性集合configMap，创建Kafka生产者对象（消息发送者对象producer），建立Kafka连接，若指定的主题不存在，则会先创建
        //      构造对象时，需要传递配置参数，一帮封装到集合中，configMap
        KafkaProducer<String, String> producer = new KafkaProducer<>(configMap);
        // TODO 准备数据（消息）,定义泛型
        //      构造对象时需要传递 【Topic主题名称】，【Key】，【Value--真正的的消息】三个参数
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<String, String>(
                    "test", "key" + i, "value" + i
            );
            // TODO 生产（发送）数据
            producer.send(record);
        }

        // TODO 关闭生产者连接
        producer.close();
    }

}
